package com.system.ad.mysql.listener;

import com.github.shyiko.mysql.binlog.event.EventType;
import com.system.ad.mysql.constant.Constant;
import com.system.ad.mysql.constant.OpType;
import com.system.ad.mysql.dto.BinlogRowData;
import com.system.ad.mysql.dto.MySqlRowData;
import com.system.ad.mysql.dto.TableTemplate;
import com.system.ad.mysql.sender.ISender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 实现增量索引
 */
@Slf4j
@Component
public class IncrementListener implements IListener {

    // 根据名称注入索引投递方式
    @Resource(name = "IndexSender")
    private ISender iSender;

    @Autowired
    private AggregationListener aggregationListener;

    @Autowired
    public IncrementListener(AggregationListener aggregationListener) {
        this.aggregationListener = aggregationListener;

    }

    /**
     * 注册需要监听的数据表
     */
    @Override
    @PostConstruct
    public void register() {
        log.info("IncrementListener register db and table info");
        Constant.table2Db.forEach((k, v) -> aggregationListener.register(v, k, this));
    }

    /**
     * 处理日志事件
     *
     * @param eventData
     */
    @Override
    public void onEvent(BinlogRowData eventData) {
        TableTemplate table = eventData.getTable();
        EventType eventType = eventData.getEventType();

        // 转换为需要投递的数据
        MySqlRowData mySqlRowData = new MySqlRowData();
        mySqlRowData.setTableName(table.getTableName());
        OpType optype = OpType.to(eventType);
        mySqlRowData.setOpType(optype);

        // 取出模板中该操作的对应字段
        List<String> fieldList = table.getOpTypeFieldSetMap().get(optype);
        if (null == fieldList) {
            log.warn("{} not support for {}", optype, table.getTableName());
            return;
        }
        // 循环更新的数据
        //填充数据
        for (Map<String, String> afterMap : eventData.getAfter()) {
            HashMap<String, String> _afterMap = new HashMap<>();
            for (Map.Entry<String, String> entry : afterMap.entrySet()) {
                String colName = entry.getKey();
                String colValue = entry.getValue();
                _afterMap.put(colName, colValue);
            }
            mySqlRowData.getFieldValueMap().add(_afterMap);
        }
        //投递消息
        iSender.sender(mySqlRowData);
    }
}
